package com.cdel.akka.agent;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import com.cdel.akka.route.WatchActor;
import com.typesafe.config.ConfigFactory;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Inbox;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.agent.Agent;
import akka.dispatch.ExecutionContexts;
import akka.dispatch.Futures;
import akka.dispatch.OnComplete;

//如果多个actor需要对同一个共享变量进行读写时，需要使用Agent。一个Agent提供了对一个变量的异步更新，当一个actor希望改变Agent的值时，它会向这个Agent
//下发一个动作。当多个actor同时改变Agent时，这些actor将会在ExecutionContext中被并发调度执行。在任意时刻，一个Agent最多只能执行一个actor，对于某一
//个线程来说，他执行actor的顺序与他的发生顺序一致，但对于不同线程来说，这些actor可能会交织在一起。
//Agent的修改可以使用send或alter。他们都可以向agent发送一个修改动作。但是send没有返回值，而alter会返回一个future对象便于跟踪agent的执行。
public class AgentDemo {

	public static Agent<Integer> counterAgent = Agent.create(0, ExecutionContexts.global());
	static ConcurrentLinkedQueue<Future<Integer>> futures = new ConcurrentLinkedQueue<>();

	public static void main(String[] args) throws Exception {
		final ActorSystem system = ActorSystem.create("agentdemo", ConfigFactory.load("samplehello.conf"));
		ActorRef[] counter = new ActorRef[10];
		for(int i=0; i<counter.length; i++){
			counter[i] = system.actorOf(Props.create(CounterActor.class), "counter_" + i);
		}
		final Inbox inbox = Inbox.create(system);
		for(int i=0; i<counter.length; i++){
			inbox.send(counter[i], 1);
			inbox.watch(counter[i]);
		}

		int closeCount = 0;
		//等待所有actor全部结束，执行完成后，我们便已经收集到了所有的future。
		while(true){
			Object msg = inbox.receive(Duration.create(1, TimeUnit.SECONDS));
			if(msg instanceof Terminated){
				closeCount++;
				if(closeCount == counter.length){
					break;
				}
			}else{
				System.out.println(msg);
			}
		}
		//将所有的future进行串行组合（使用sequence方法），构造了一个整体的future，并为他创建onComplete回调函数。在所有的agent操作执行完成后，
		//onComplete方法就会被调用。
		Futures.sequence(futures, system.dispatcher()).onComplete(new OnComplete<Iterable<Integer>>() {
			@Override
			public void onComplete(Throwable arg0, Iterable<Integer> arg1) throws Throwable {
				System.out.println("counterAgent=" + counterAgent.get());
				system.shutdown();
			}
		}, system.dispatcher());
	}

}
